package cc.eddic.practice.concurrency;

import lombok.extern.slf4j.Slf4j;
import lombok.val;

import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import static cc.eddic.practice.concurrency.ExecutorServiceUtil.shutdown;

@Slf4j
public class LogParserTyq0811Impl implements LogParser {

    LogItemParseTyq0811Impl logItemParserTyq0811 = new LogItemParseTyq0811Impl();

    @Override
    public Map<String, List<LogItem>> parse(String folderName) {
        Map<String, List<LogItem>> resultMap = new HashMap<>();

        List<Future<Map<String, List<LogItem>>>> futures;
        ExecutorService executorService = Executors.newFixedThreadPool(17);
        val p = Paths.get(folderName);
        try (val walk = Files.walk(p)) {
            futures = walk.filter(Files::isRegularFile)
                    .filter(path -> path.getFileName().toString().endsWith(".txt"))
                    .map(path -> executorService.submit(() -> parse(path)))
                    .collect(Collectors.toList());
        } catch (IOException e) {
            throw new RuntimeException(e);
        }

        if (log.isDebugEnabled()) log.debug("{} future is running", futures.size());
        allOf(futures);

        if (log.isDebugEnabled()) log.debug("get Map");
        futures.stream().parallel().map(this::getFuture).filter(Objects::nonNull)
                .forEach(map ->
                        {
                            if (log.isDebugEnabled()) log.debug("{} 即将put", map.keySet());
                            resultMap.putAll(map);
                            if (log.isDebugEnabled()) log.debug("{} 已经put", map.keySet());
                        }
                );
        shutdown(executorService);
        return resultMap;
    }

    private Map<String, List<LogItem>> getFuture(Future<Map<String, List<LogItem>>> future) {
        try {
            return future.get();
        } catch (InterruptedException | ExecutionException e) {
            throw new RuntimeException(e);
        }
    }

    private static void sleep(int millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            log.error(e.getMessage());
        }
    }

    private static <T> void allOf(final List<Future<T>> futures) {
        while (true) {
            val b = futures.stream().anyMatch(x -> !x.isDone());
            if (!b) break;

            sleep(10);
            if (log.isDebugEnabled()) log.debug("{} futures done", futures.stream().filter(Future::isDone).count());
        }
    }

    //解析每一个log文件
    private Map<String, List<LogItem>> parse(Path path) {
        File file = path.toFile();
        Map<String, List<LogItem>> map = new HashMap<>();
        if (log.isDebugEnabled()) log.debug("{}", file.getName());
        List<LogItem> logItemList = readList(file.getPath()).stream().map(filestring -> logItemParserTyq0811.parse(filestring)).filter(Optional::isPresent).map(Optional::get).collect(Collectors.toList());
        map.put(file.getName(), logItemList);
        return map;
    }

    //parse：读log文件
    public List<String> readList(String textFileName) {
        return read(textFileName).collect(Collectors.toList());
    }

    //parse：读每一行
    public Stream<String> read(String textFileName) {
        try {
            val p = Paths.get(textFileName);
            return Files.lines(p);
        } catch (IOException e) {
            if (log.isDebugEnabled()) log.debug("{}", textFileName);
            return null;
        }
    }
}
